#include "spidernode.h"

#include "master.h"
#include "define.h"
#include "log.h"
#include "util.h"

#include <string.h>

static void _OnSpiderNodeClose(uv_handle_t *handle)
{
    LOG_DEBUG("_OnSpiderNodeClose");
    if (handle->data)
    {
        return;
    }
    SpiderNode *pSpiderNode = (SpiderNode *)handle->data;
    delete pSpiderNode;
}

static void _OnSpiderNodeCheck(uv_timer_t *handle)
{
    SpiderNode *pSpiderNode = (SpiderNode *)uv_handle_get_data((uv_handle_t *)handle);
    if (pSpiderNode == NULL)
    {
        uv_timer_stop(handle);
        return;
    }

    pSpiderNode->OnSpiderNodeCheck();
}

static void _OnAllocBuff(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
    SpiderNode *pSpiderNode = (SpiderNode *)uv_handle_get_data(handle);

    buf->base = pSpiderNode->m_recvBuf + pSpiderNode->m_recvLen;
    buf->len = TCP_BUFF_SIZE - pSpiderNode->m_recvLen;
    return;
}

static void _OnReadData(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf)
{
    LOG_DEBUG("_OnReadData");
    SpiderNode *pSpiderNode = (SpiderNode *)uv_handle_get_data((uv_handle_t *)stream);
    if (nread >= 0 && pSpiderNode != NULL)
    {
        pSpiderNode->HandReaddata(nread);
        return;
    }
    pSpiderNode->Close(true);
    return;
}
static void _OnWriteData(uv_write_t *req, int status)
{
    LOG_DEBUG("_OnWriteData");
    SpiderNode *pSpiderNode = (SpiderNode *)uv_handle_get_data((uv_handle_t *)req);
    if (pSpiderNode == NULL)
    {
        free(req);
        return;
    }
    if (status != 0)
    {
        LOG_ERROR("SpiderNode [%s] Send Message Failed", pSpiderNode->m_uuid.c_str());
        pSpiderNode->Close(true);
    }
    pSpiderNode->m_sendLen = 0;
    free(req);
    return;
}

bool SpiderNode::Init(Master *master, uv_stream_t *pServer, uint32_t interval)
{
    m_master = master;
    m_interval = interval * 3;
    uv_tcp_init(m_master->GetLoop(), &m_connect);
    if (uv_accept(pServer, (uv_stream_t *)&m_connect) == 0)
    {
        struct sockaddr addr;
        int addrlen = sizeof(struct sockaddr);
        // 获取客户端IP地址
        if (uv_tcp_getpeername(&m_connect, (struct sockaddr *)&addr, &addrlen) == 0)
        {
            if (addr.sa_family == AF_INET)
            {
                struct sockaddr_in const *ipv4 = (struct sockaddr_in *)&addr;
                m_ipaddr = ipv4->sin_addr.s_addr;
            }
            else
            {
                LOG_INFO("SpiderNode not support ipv6");
                Close(false);
                return false;
            }
        }
        uv_timer_init(m_master->GetLoop(), &m_spiderTimer);
        uv_handle_set_data((uv_handle_t *)&m_spiderTimer, this);
        uv_handle_set_data((uv_handle_t *)&m_connect, this);
        m_lastRecvTick = uv_now(m_master->GetLoop());
        uv_timer_start(&m_spiderTimer, _OnSpiderNodeCheck, 0, g_timerRepeat);
        uv_read_start((uv_stream_t *)&m_connect, _OnAllocBuff, _OnReadData);
        m_isInit = true;
        LOG_DEBUG("SpiderNode::Init");
        return true;
    }
    Close(false);
    return false;
}

void SpiderNode::HandReaddata(ssize_t nread)
{
    LOG_DEBUG("SpiderNode::HandReaddata: %d", nread);
    m_lastRecvTick = uv_now(m_master->GetLoop());
    m_recvLen += nread;
    char *recvPos = m_recvBuf;
    do
    {
        if (m_recvLen < sizeof(MessageHeader))
        {
            if (recvPos != m_recvBuf)
            {
                memmove(m_recvBuf, recvPos, m_recvLen);
                recvPos = m_recvBuf;
            }
            return;
        }
        MessageHeader *header = (MessageHeader *)recvPos;
        switch (header->type)
        {
        case MESSAGE_REGISTER_REQ:
            if (m_recvLen < sizeof(MessageRegister))
            {
                if (recvPos != m_recvBuf)
                {
                    memmove(m_recvBuf, recvPos, m_recvLen);
                    recvPos = m_recvBuf;
                }
                return;
            }
            LOG_DEBUG("Spider Node DoRegister");
            if (!DoRegister((MessageRegister *)recvPos)) 
            {
                LOG_ERROR("Spider Node DoRegister Failed");
                Close(true);
                return;
            }
            recvPos += sizeof(MessageRegister);
            m_recvLen -= sizeof(MessageRegister);
            SendMessageRegisterResp();
            break;
        case MESSAGE_HEARTBEAT_REQ:
            LOG_DEBUG("Spider Node Heartbeat");
            SendHeartbeatResp((MessageHeartbeat *)recvPos);
            recvPos += sizeof(MessageHeartbeat);
            m_recvLen -= sizeof(MessageHeartbeat);
            break;
        default:
            LOG_INFO("Unsupported MsgType %d", header->type);
            Close(true);
            return;
            break;
        }
    } while (true);
}

bool SpiderNode::DoRegister(MessageRegister *msg)
{
    LOG_DEBUG("DoRegister");
    std::string msg_udid = msg->udid;
    std::string msg_nonce = msg->nonce;
    std::string msg_cipher = msg->cipher;
    std::string cipher = md5(msg_udid + msg_nonce + g_Cipher);
    LOG_DEBUG("DoRegister msg_udid %s", msg_udid.c_str());
    LOG_DEBUG("DoRegister msg_nonce %s", msg_nonce.c_str());
    LOG_DEBUG("DoRegister msg_cipher %s", msg_cipher.c_str());
    LOG_DEBUG("DoRegister cipher %s", cipher.c_str());
    if (cipher.compare(msg_cipher) != 0) {
        LOG_DEBUG("Spider Node [%s] Register cipher is error", msg_udid.c_str());
        return false;
    }
    m_uuid = msg->udid;
    if (!m_master->AddSpiderNodeCache(this)) {
        LOG_ERROR("Spider Node [%s] Register Add Cache Failed", msg_udid.c_str());
        return false;
    }
    uint8_t *ipStr = (uint8_t *)&m_ipaddr;
    LOG_INFO("Spider Node [%s] Addr[%u.%u.%u.%u] Online", m_uuid.c_str(), ipStr[0], ipStr[1], ipStr[2], ipStr[3]);
    return true;
}

void SpiderNode::Close(bool cache)
{
    LOG_DEBUG("SpiderNode Close");
    if (!m_uuid.empty()) {
        LOG_INFO("Spider Node [%s] Offline", m_uuid.c_str());
    }
    
    if (m_isInit) {
        if (cache) {
            m_master->DelSpiderNodeCache(this);
        }        
        uv_timer_stop(&m_spiderTimer);        
    }
    if (!uv_is_closing((uv_handle_t *)&m_connect)) {
       uv_close((uv_handle_t *)&m_connect, _OnSpiderNodeClose);
    }
}

void SpiderNode::OnSpiderNodeCheck()
{
    uint64_t nowTime = uv_now(m_master->GetLoop());
    if (m_lastRecvTick + m_interval < nowTime)
    {
        LOG_INFO("SpiderNode [%s] Heartbeat Timeout", m_uuid.c_str());
        Close(true);
    }
}

void SpiderNode::SendMessageRegisterResp()
{
    uv_write_t *writeReq = (uv_write_t *)calloc(1, sizeof(uv_write_t));
    if (!writeReq)
    {
        LOG_ERROR("SpiderNode [%s] calloc write Req Failed", m_uuid.c_str());
        Close(true);
        return;
    }
    uv_handle_set_data((uv_handle_t *)writeReq, (void *)this);
    MessageRegisterResp resp;
    resp.header.type = MESSAGE_REGISTER_RESP;
    resp.code = 0;
    resp.commandTimeout = m_master->m_commandTimeout;
    resp.idleTimeout = m_master->m_idleTimeout;
    uv_buf_t buf = uv_buf_init((char *)&resp, sizeof(MessageRegisterResp));
    if (uv_write(writeReq, (uv_stream_t *)&m_connect, &buf, 1, _OnWriteData) < 0)
    {
        LOG_ERROR("SpiderNode [%s] write Req Failed", m_uuid.c_str());
        free(writeReq);
        Close(true);
    }
}

void SpiderNode::SendHeartbeatResp(MessageHeartbeat *resp) 
{
    uv_write_t *writeReq = (uv_write_t *)calloc(1, sizeof(uv_write_t));
    if (!writeReq)
    {
        LOG_ERROR("SpiderNode [%s] calloc write Req Failed", m_uuid.c_str());
        Close(true);
        return;
    }
    uv_handle_set_data((uv_handle_t *)writeReq, (void *)this);
    MessageHeartbeat *msg = (MessageHeartbeat *)(m_sendBuf + m_sendLen);
    msg->header.type = MESSAGE_HEARTBEAT_RESP;
    msg->timestamp = resp->timestamp;
    m_sendLen += sizeof(MessageHeartbeat);
    uv_buf_t buf = uv_buf_init((char *)m_sendBuf, m_sendLen);
    if (uv_write(writeReq, (uv_stream_t *)&m_connect, &buf, 1, _OnWriteData) < 0)
    {
        LOG_ERROR("SpiderNode [%s] write Req Failed", m_uuid.c_str());
        free(writeReq);
        Close(true);
    }
}

void SpiderNode::SendTask(char *task, uint16_t taskLen) 
{
    LOG_DEBUG("SendTask %u", taskLen);
    if (m_sendLen + taskLen > TCP_BUFF_SIZE) {
        LOG_ERROR("SpiderNode [%s] SendBuff Is Full", m_uuid.c_str());
        return;
    }
    uv_write_t *writeReq = (uv_write_t *)calloc(1, sizeof(uv_write_t));
    if (!writeReq)
    {
        LOG_ERROR("SpiderNode [%s] calloc write Req Failed", m_uuid.c_str());
        Close(true);
        return;
    }
    uv_handle_set_data((uv_handle_t *)writeReq, (void *)this);    
    memcpy(m_sendBuf + m_sendLen, task, taskLen);

    m_sendLen += taskLen;
    uv_buf_t buf = uv_buf_init((char *)m_sendBuf, m_sendLen);
    if (uv_write(writeReq, (uv_stream_t *)&m_connect, &buf, 1, _OnWriteData) < 0)
    {
        LOG_ERROR("SpiderNode [%s] write Req Failed", m_uuid.c_str());
        free(writeReq);
        Close(true);
    }
}